package com.bs.flinktest.vo;

import cn.hutool.core.lang.Snowflake;
import lombok.Data;
import lombok.extern.log4j.Log4j2;
import org.apache.flink.api.common.JobID;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.io.Serializable;
import java.nio.charset.StandardCharsets;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Callable;

/**
 * @author bingshao
 * @date 2023/2/1
 **/
@Data
@Log4j2
public class TestDataSource implements Serializable, Callable<JobClient> {
    private static final long serialVersionUID = -2342825989707527491L;

    public static List<JobClient> jobClientList = new LinkedList<>();


    private String hostName;

    private Integer port;

    private String username;

    private String password;

    private String databaseName;

    private String tableName;

    private String targetTableName;

    private String targetUrl;

    private String targetUsername;

    private String targetPassword;

    private String driver;


    @Override
    public JobClient call() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60 * 1000L);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30 * 1000L);
        env.getCheckpointConfig().setCheckpointTimeout(60 * 2 * 1000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//        env.getCheckpointConfig().setCheckpointStorage("file:///checkpoint");
        StreamTableEnvironment ste = StreamTableEnvironment.create(env);
        String sql = "create table `mysql1` ( id bigint NOT NULL COMMENT '主键',user_id bigint,name VARCHAR(20) CHARACTER SET `UTF-8` COMMENT '姓名',PRIMARY KEY (id) NOT ENFORCED) ";
        ste.executeSql(
                sql +
                        " WITH (" +
                        " 'scan.incremental.snapshot.enabled' = 'false'," +
//                        " 'scan.startup.mode' = 'initial', "+
                        " 'connector' = 'mysql-cdc'," +
                        " 'hostname' = '" + this.hostName + "'," +
                        " 'port' = '" + this.port + "'," +
                        " 'username' = '" + this.username + "'," +
                        " 'password' = '" + this.password + "'," +
                        " 'database-name' = '" + this.databaseName + "'," +
                        " 'table-name' = '" + this.tableName + "'" +
                        ")"
        );

//        ste.executeSql(
//                "create table `mysql1" + "`" + "( id bigint NOT NULL COMMENT '主键',name VARCHAR," +
//                        "    PRIMARY KEY (id) NOT ENFORCED " +
//                        ") WITH (" +
//                        " 'scan.incremental.snapshot.enabled' = 'false'," +
////                        " 'scan.startup.mode' = 'initial', "+
//                        " 'connector' = 'mysql-cdc'," +
//                        " 'hostname' = '" + this.hostName + "'," +
//                        " 'port' = '" + this.port + "'," +
//                        " 'username' = '" + this.username + "'," +
//                        " 'password' = '" + this.password + "'," +
//                        " 'database-name' = '" + this.databaseName + "'," +
//                        " 'table-name' = '" + this.tableName + "'" +
//                        ")"
//        );
//            ste.executeSql(
//                    "create table mysqlorder( id bigint,user_id bigint,product_name VARCHAR," +
//                            "    PRIMARY KEY (id) NOT ENFORCED" +
//                            ") WITH (" +
//                            " 'scan.incremental.snapshot.enabled' = 'false'," +
//                            " 'connector' = 'mysql-cdc'," +
//                            " 'hostname' = '110.42.233.197'," +
//                            " 'port' = '3306'," +
//                            " 'username' = 'root'," +
//                            " 'password' = 'bs1354361838'," +
//                            " 'database-name' = 'bingshao'," +
//                            " 'table-name' = 'order'" +
//                            ")"
//            );

        ste.executeSql(
                "create table `mysqlsink" + "`" + "(" +
                        "id bigint,user_id bigint,name VARCHAR,status CHAR," +
                        "    PRIMARY KEY (id) NOT ENFORCED" +
                        ") WITH (" +
                        "'connector' = 'jdbc'," +
                        "'url' = '" + this.targetUrl + "'," +
                        "'username' = '" + this.targetUsername + "'," +
                        "'password' = '" + this.targetPassword + "'," +
                        "'table-name' = '" + this.targetTableName + "'," +
                        "'driver' = '" + this.driver + "'" +
                        ")"
        );

        String s1 = "id,user_id,name,status";

        String s2 = "select id,user_id,name";

        Object o1;


//        s2 = s2.substring(s2.indexOf("select") + 6).trim();

        TableResult tableResult = ste.executeSql(
//                "insert into mysqlsink(" + s1 + ") " + "select cast(CONCAT(cast(UNIX_TIMESTAMP() as CHAR),cast(FLOOR(1 + (RAND() * 100001)) as CHAR)) as BIGINT) AS id"  + ", " + s2
                "insert into mysqlsink(id,`" + s1 + "`)"  + "select " + new Snowflake(1, 1).nextId() + s2
        );
        JobClient jobClient = tableResult.getJobClient().get();
//        JobID jobID = tableResult.getJobClient().get().getJobID();
        jobClientList.add(jobClient);
        return jobClient;
    }
}
